/**
 * @author      Xu Meng <meng.xu01@sap.com>
 * @version     0.1                 
 * @since       2012-06-18
 */

package sap.feed.java.client;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.json.JSONException;
import org.json.JSONObject;

import sap.socketio.java.client.IOSocket;
import sap.socketio.java.client.MessageCallback;

public class SAPFeed {
	
	private IOSocket socket;
	private SAPFeedInterface client;
	private ExecutorService callbackExecutor;
	private ExecutorService serviceExecutor;
	private Hashtable<Integer, SAPFeedLogEntry> activityLog;
	private boolean connected;
	private boolean loggedIn;
	
	
	/**
	 * Constructor of SAPFeed.
	 *
	 * On creation of SAPFeed instances, connection to server will be auto-established if network condition permits.
	 * On successful connection to server, the program will authenticate user with userID and password provided.
	 * A reference to the client program established to handle SAPFeed callback functions. 
	 * The activity log is initialized and the login activity is logged.
	 * Upon successful login, call onSAPFeedLogin() in the client program.
	 * In case of error, call onSAPFeedError() in the client program with specific error code.
	 * 
	 *
	 * @param  serverAddress the URI of the SAPFeed server.
	 * @param  cientProgram an instance of a class that has implemented SAPFeedInterface
	 * @param  userID a unique user ID to represent each user of SAPFeed service
	 * @param  password a password associated with each user ID for authenticate purpose
	 * @return a SAPFeed Object.
	 */	
	public SAPFeed(String serverAddress, SAPFeedInterface clientProgram, final int userID, final String password){
		connected = false;
		loggedIn = false;
		client = clientProgram;
		
		callbackExecutor = Executors.newCachedThreadPool();
		serviceExecutor = Executors.newCachedThreadPool();
		
		activityLog = new Hashtable<Integer, SAPFeedLogEntry>();	
		
		socket = new IOSocket(serverAddress, new MessageCallback(){

			@Override
			public void on(String event, JSONObject... data){
				
				try {
					
					if(data[0].has("activityID")){
					
						String responseMessage = data[0].getString("message");					
						String responseCode = data[0].getString("code");
						int activityID = data[0].getInt("activityID");
					
						activityLog.get(activityID).setResponse(System.currentTimeMillis(), responseMessage, responseCode);
						
						if(responseCode.charAt(0)=='E'){
							if(loggedIn){
								logout();
							}

							client.onSAPFeedError(responseCode, responseMessage);
						}
						
					}
					else{
						int activityID = activityLog.size();
						activityLog.put(activityID, new SAPFeedLogEntry(event, System.currentTimeMillis(),false));
					}
					
				} catch (JSONException e) {
					client.onSAPFeedError("E902", "Error Decoding Server Message");
				}				
				
				if(event.equals("login_success")){
					loggedIn = true;
					callbackExecutor.execute(new Runnable(){
					
						@Override
						public void run() {
							// TODO Auto-generated method stub
							client.onSAPFeedLogin();
						}
						
					});
					
				}

				else if(event.equals("logout_success")){
					loggedIn = false;
					disconnect();
					callbackExecutor.execute(new Runnable(){

						@Override
						public void run() {
							client.onSAPFeedLogout();
						}
						
					});						

				}

				else if(event.equals("publish_success")){

					callbackExecutor.execute(new Runnable(){

						@Override
						public void run() {
							client.onSAPFeedPublish();
						}
						
					});						
				}

				else if(event.equals("listen_success")){
					callbackExecutor.execute(new Runnable(){

						@Override
						public void run() {
							client.onSAPFeedListen();
						}
						
					});					
				}

				else if(event.equals("data_transfer")){
					Object dataTmp;
					int datasetIDTmp;
					int publisherIDTmp;
					try{
						dataTmp = data[0].get("data");
						datasetIDTmp = data[0].getInt("dataset");
						publisherIDTmp = data[0].getInt("publisher");
					} catch (JSONException e) {
						throw new SAPFeedRuntimeException("E902", event+": "+"Error Decoding Server Message");
					}					
					final int datasetPass = datasetIDTmp;
					final int publisherIDPass = publisherIDTmp;
					final Object dataPass = dataTmp;
					callbackExecutor.execute(new Runnable(){

						@Override
						public void run() {
							client.onSAPFeedReceive(publisherIDPass,datasetPass,dataPass);
						}
						
					});	
			
				}
			}


			@Override
			public void onMessage(String message) {
				// TODO Auto-generated method stub
				
			}

			@Override
			public void onMessage(JSONObject json) {
				// TODO Auto-generated method stub
				
			}

			@Override
			public void onConnect() {
				// TODO Auto-generated method stub
				connected = true;
				int activityID = activityLog.size();
				activityLog.put(activityID, new SAPFeedLogEntry("connect_confirmation",System.currentTimeMillis(),false));
				
				login(userID, password);
				
			}

			@Override
			public void onDisconnect() {
				// TODO Auto-generated method stub
				connected = false;
				int activityID = activityLog.size();
				activityLog.put(activityID, new SAPFeedLogEntry("disconnect_confirmation",System.currentTimeMillis(),false));					
			}

		});
		
		connect();
		
	}
	
	private void connect() {
		
		Runnable task = new Runnable(){

			@Override
			public void run() {
				if(!connected){
					try {
						socket.connect();
					} catch (IOException e) {
						throw new SAPFeedRuntimeException("E001",e.getMessage()); 
					}
				}else{
					throw new SAPFeedRuntimeException("E005","Already in connection");
				}				
			}
			
		};
		
		Future<?> future = serviceExecutor.submit(task); 
		
		try {
			future.get();
		} catch (InterruptedException e1) {
			client.onSAPFeedError("E903", "Error Executing Asynchronous Tasks");
		} catch (ExecutionException e1) {
			SAPFeedRuntimeException ex = (SAPFeedRuntimeException)e1.getCause();
			client.onSAPFeedError(ex.getErrorCode(), ex.getMessage());
		}
	}
	
	private void disconnect() {
		
		Runnable task = new Runnable(){
			@Override
			public void run() {
				if(connected){
					if(!loggedIn){
						socket.disconnect();
					}else{
						throw new SAPFeedRuntimeException("E004","Disconnect without Logout");
					}
				}else{
					throw new SAPFeedRuntimeException("E001","Connection Not Established");
				}
			}
			
		};
		
		Future<?> future = serviceExecutor.submit(task); 
		
		try {
			future.get();
		} catch (InterruptedException e1) {
			client.onSAPFeedError("E903", "Error Executing Asynchronous Tasks");
		} catch (ExecutionException e1) {
			SAPFeedRuntimeException ex = (SAPFeedRuntimeException)e1.getCause();
			client.onSAPFeedError(ex.getErrorCode(), ex.getMessage());
		}		
	}
	
	private void login(final int userID, final String password){
		
		Runnable task = new Runnable(){

			@Override
			public void run() {

				final int activityID = activityLog.size();
				activityLog.put(activityID, new SAPFeedLogEntry("login",System.currentTimeMillis(),true));
				
				if(!connected){
					callbackExecutor.execute(new Runnable(){

						@Override
						public void run() {
							String message = "Connection Not Established";
							activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E001");
							throw new SAPFeedRuntimeException("E001",message);
						}
						
					});

				}
				else if(loggedIn){
					callbackExecutor.execute(new Runnable(){

						@Override
						public void run() {
							String message = "Duplicated Login";
							activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E101");							
							throw new SAPFeedRuntimeException("E101",message);
						}
						
					});
				}
				else{
					JSONObject loginInfo = new JSONObject();
					
					try {
						loginInfo.put("activityID", activityID);
						loginInfo.put("userID",userID);
						loginInfo.put("password", password);
						socket.emit("login_request", loginInfo);
					
					} catch (JSONException e) {
						callbackExecutor.execute(new Runnable(){

							@Override
							public void run() {			
								String message = "Illegal Form of Information/Data";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E901");									
								throw new SAPFeedRuntimeException("E901",message);
							}
						});
					} catch (IOException e) {
						callbackExecutor.execute(new Runnable(){

							@Override
							public void run() {			
								String message = "Exception in Server-Client Communication";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E902");								
								throw new SAPFeedRuntimeException("E902",message);
							}
						});						
					}				
				}			
			}
			
		};
		
		Future<?> future = serviceExecutor.submit(task); 
		
		try {
			future.get();
		} catch (InterruptedException e1) {
			client.onSAPFeedError("E903", "Error Executing Asynchronous Tasks");
		} catch (ExecutionException e1) {
			SAPFeedRuntimeException ex = (SAPFeedRuntimeException)e1.getCause();
			client.onSAPFeedError(ex.getErrorCode(), ex.getMessage());
		}		

	}

	/**
	 * Logout from SAPFeed service with current account.
	 *
	 * On successful logout, call onSAPFeedLogout() in the client program,
	 * On unsuccessful logout, call onSAPFeedLogoutError() in the client program with the error message
	 *
	 */		
	public void logout(){
		
		Runnable task = new Runnable(){
			@Override
			public void run() {
				final int activityID = activityLog.size();
				activityLog.put(activityID, new SAPFeedLogEntry("logout",System.currentTimeMillis(),true));						
				
				if(!connected){
					callbackExecutor.execute(new Runnable(){

						@Override
						public void run() {
							String message = "Connection Not Established";
							activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E001");							
							throw new SAPFeedRuntimeException("E001",message);
						}
						
					});

				}
				else if(loggedIn){
				
					JSONObject logoutInfo = new JSONObject();
					try {
						logoutInfo.put("activityID", activityID);
						socket.emit("logout_request", logoutInfo);
					} catch (JSONException e1) {
						// TODO Auto-generated catch block
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Illegal Form of Information/Data";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E901");								
								throw new SAPFeedRuntimeException("E901",message);
							}
						});
					}
					catch (IOException e) {
						// TODO Auto-generated catch block

						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Exception in Server-Client Communication";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E002");								
								throw new SAPFeedRuntimeException("E002",message);
							}
						});
						
					}		
				}
				else{
					callbackExecutor.execute(new Runnable(){
						public void run(){
							String message = "Action without Login";
							activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E103");								
							throw new SAPFeedRuntimeException("E103",message);
						}
					});
				}	
			}
			
		};
		
		Future<?> future = serviceExecutor.submit(task); 
		
		try {
			future.get();
		} catch (InterruptedException e1) {
			client.onSAPFeedError("E903", "Error Executing Asynchronous Tasks");
		} catch (ExecutionException e1) {
			SAPFeedRuntimeException ex = (SAPFeedRuntimeException)e1.getCause();
			client.onSAPFeedError(ex.getErrorCode(), ex.getMessage());
		}			

	}	
	
	/**
	 * Publish data to datasets (indicated by a list of data set IDs) on SAPFeed service.
	 *
	 * On successful publish, call onSAPFeedPublish() in the client program
	 * On unsuccessful publish, call onSAPFeedPublishError() in the client program with the error message
	 *
	 * @param datasetList a list of dataset IDs to publish to, in ArrayList<Integer> format.
	 * @param data an Object which contains the data to be published
	 * 
	 */		
	public void publish(final ArrayList<Integer> datasetList, final Object data){
		
		Runnable task = new Runnable(){
			@Override
			public void run() {
				final int activityID = activityLog.size();
				activityLog.put(activityID, new SAPFeedLogEntry("publish",System.currentTimeMillis(),true));							
				
				if(connected&&loggedIn){
				
					JSONObject publishInfo = new JSONObject();
			
					
					try {
						publishInfo.put("activityID",activityID);
						publishInfo.put("datasetList",datasetList);
						publishInfo.put("data",data);
					
						socket.emit("publish", publishInfo);
						
					} catch (JSONException e) {
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Illegal Form of Information/Data";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E901");									
								throw new SAPFeedRuntimeException("E901",message);
							}
						});

					} catch (IOException e) {
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Exception in Server-Client Communication";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E002");										
								throw new SAPFeedRuntimeException("E002",message);
							}
						});						
						
					}
				}
				else{
					if(!connected){
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Connection Not Established";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E001");								
								throw new SAPFeedRuntimeException("E001",message);
							}
						});						
						
						
					}
					else if(!loggedIn){
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Action without Login";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E103");									
								throw new SAPFeedRuntimeException("E103",message);
							}
						});		
						
					}
				}
			}
			
		};
		
		Future<?> future = serviceExecutor.submit(task); 
		
		try {
			future.get();
		} catch (InterruptedException e1) {
			client.onSAPFeedError("E903", "Error Executing Asynchronous Tasks");
		} catch (ExecutionException e1) {
			SAPFeedRuntimeException ex = (SAPFeedRuntimeException)e1.getCause();
			client.onSAPFeedError(ex.getErrorCode(), ex.getMessage());
		}			
	}	


	/**
	 * Listen to (start receiving data from) datasets (indicated by a list of data set IDs) on SAPFeed service.
	 *
	 * On successful listen, call onSAPFeedListen() in the client program
	 * On unsuccessful listen, call onSAPFeedListenError() in the client program with the error message
	 *
	 * @param datasetList a list of dataset IDs to listen to, in ArrayList<Integer> format.
	 * 
	 */	
	public void listen(final ArrayList<Integer> datasetList){
		
		
		Runnable task = new Runnable(){

			@Override
			public void run() {
				final int activityID = activityLog.size();
				activityLog.put(activityID, new SAPFeedLogEntry("listen",System.currentTimeMillis(),true));				
				
				if(connected&&loggedIn){
					
					JSONObject listenInfo = new JSONObject();
			
					
					try {
						listenInfo.put("activityID",activityID);
						listenInfo.put("datasetList",datasetList);
					
						socket.emit("listen", listenInfo);
						
					} catch (JSONException e) {
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Illegal Form of Information/Data";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E901");									
								throw new SAPFeedRuntimeException("E901",message);
							}
						});

					} catch (IOException e) {
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Exception in Server-Client Communication";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E002");									
								throw new SAPFeedRuntimeException("E002",message);
							}
						});						
						
					}
				}
				else{
					if(!connected){
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Connection Not Established";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E001");								
								throw new SAPFeedRuntimeException("E001",message);
							}
						});						
						
						
					}
					else if(!loggedIn){
						callbackExecutor.execute(new Runnable(){
							public void run(){
								String message = "Action without Login";
								activityLog.get(activityID).setResponse(System.currentTimeMillis(), message, "E103");
								throw new SAPFeedRuntimeException("E103",message);
							}
						});		
						
					}
				}				
			}
			
		};
		
		Future<?> future = serviceExecutor.submit(task); 
		
		try {
			future.get();
		} catch (InterruptedException e1) {
			client.onSAPFeedError("E903", "Error Executing Asynchronous Tasks");
		} catch (ExecutionException e1) {
			SAPFeedRuntimeException ex = (SAPFeedRuntimeException)e1.getCause();
			client.onSAPFeedError(ex.getErrorCode(), ex.getMessage());
		}		

	}

	/**
	 * Get the activity log for in this session.
	 *
	 * The entries in the log are instances of SAPFeedLogEntry.
	 * 
	 * @return Hashtable<Integer, SAPFeedLogEntry> activity log in Hashtable format
	 * 
	 */
	public Hashtable<Integer, SAPFeedLogEntry> getActivityLog(){
		return activityLog;
	}	

	private class SAPFeedRuntimeException extends RuntimeException{

		private static final long serialVersionUID = 2098111181516519673L;
		
		private String errorCode;
		public SAPFeedRuntimeException (String errorCode, String message){
			super(message);
			this.errorCode = errorCode;
		}
		public String getErrorCode(){
			return errorCode;
		}
	}

	
}
